package gouroutine_pool

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

// 通知main函数任务递交到协程池的任务全部完成
var stopMain = make(chan struct{}, 1)

type GoroutinePool struct {
	dataChannel        chan *Task     // 对外提供的任务入口,即任务队列
	maxWorkNum         int            // 最大协程数
	strategy           RejectStrategy // 队列满之后的策略
	closed             int32          // 池是否已被关闭
	recordCompleteTask bool           // 是否需要保存任务执行结果
	endChannel         []*Task        // 已经完成了的任务
	taskFactory        taskFactory
}

var lock sync.Mutex

// 初始化默认携程池
func NewDefaultGoroutinePool(cap int) *GoroutinePool {
	res := &GoroutinePool{
		dataChannel:        make(chan *Task, 1),
		maxWorkNum:         cap,
		strategy:           &DefaultRejectStrategy{}, // 使用默认的策略
		endChannel:         make([]*Task, 1),
		recordCompleteTask: true,
	}
	return res
}

// 指定了拒绝策略和task工厂
func NewFactoryPool(cap int, factory taskFactory, strategy RejectStrategy) *GoroutinePool {
	res := &GoroutinePool{
		dataChannel: make(chan *Task, 1),
		maxWorkNum:  cap,
		strategy:    strategy, // 使用默认的策略
		endChannel:  make([]*Task, 1),
		taskFactory: factory,
	}
	return res
}

// 携程池创建一个worker并开始工作
func (p *GoroutinePool) worker(id int) {
	for task := range p.dataChannel { // 通道保证一个task只能被一个协程处理
		if task.state == NEW {
			// 执行任务
			go task.Exec() // 这里需要新起一个携程去处理任务，worker协程负责阻塞等待任务结果或者是超时处理
			t := time.Duration(task.timeOut) * task.timeUnit
			select {
			case <-time.After(t):
				task.cancel()
				task.state = TIMEOUT
				if task.callBack != nil {
					_, _ = task.callBack(task.param) //失败了则进行回调
				}
			case result := <-task.resultChan: // 在n秒后还是收不到结果数据，说明任务超时了,GetResult方法返回值是一个通道
				// 保存已完成的任务
				if p.recordCompleteTask {
					p.endChannel = append(p.endChannel, task)
				}
				fmt.Println("workId：", id, " 执行任务", task.id, "完成，执行结果为: ", result)
			}
		}
	}
}

func (p *GoroutinePool) Init() {
	// 根据携程池的数量初始化n个worker
	for i := 0; i < p.maxWorkNum; i++ {
		go p.worker(i)
	}
}

// 向协程池中增加一个任务
func (p *GoroutinePool) AddTask(t *Task) error {
	// 如果协程池已经关闭，或者池已经满了,则调用拒绝策略
	if p.IsClose() || p.IsFull() {
		callbackRes, err := p.strategy.Reject(p, t)
		fmt.Println(callbackRes)
		fmt.Println(err)
		return err
	} else {
		p.dataChannel <- t
	}
	return nil
}

// 通过工厂添加任务
func (p *GoroutinePool) AddTaskByFactory(run runFunc, param interface{}, timeOut int, timeUnit time.Duration, callBack runFunc, mata string) error {
	task := p.taskFactory.NewTask(run, param, timeOut, timeUnit, callBack, mata)
	return p.AddTask(task)
}

// 关闭携程池
func (p *GoroutinePool) Close() {
	if GoroutineOpen == atomic.LoadInt32(&p.closed) { // 避免并发修改
		//lock.Lock()
		fmt.Println("close channel")
		close(p.dataChannel)
		atomic.StoreInt32(&p.closed, GoroutineClose)
		stopMain <- struct{}{} // 使用空结构体当作信号量
		//lock.Unlock()
	}
}

// 判断携程池是否已经被关闭了
func (p *GoroutinePool) IsClose() bool {
	return atomic.LoadInt32(&p.closed) == GoroutineClose
}

// 判断携程池是否已经满了
func (p *GoroutinePool) IsFull() bool {
	return len(p.dataChannel) == p.maxWorkNum
}

// 获取已经完成的任务
func (p *GoroutinePool) CompleteTask() []*Task {
	return p.endChannel
}

func (p *GoroutinePool) CancelTask() {
	p.closed = GoroutineClose
}

func (p *GoroutinePool) StopChan() *chan struct{} {
	return &stopMain
}
